package com.mango.ch03;

import java.io.IOException;
import java.net.URISyntaxException;
//import java.util.Collections;
//import java.util.Iterator;
import java.util.List;
//import java.util.SortedMap;
//import java.util.TreeMap;
//import java.util.Map.Entry;

import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
//import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;

import com.mango.HDFSTools.HDFSAPI;

import scala.Tuple2;

/**
 * 
 * @author Mango
 *
 *         2018年8月1日-下午3:27:22
 */
public class Spark_Top10NonUnique {
	private static Path inputPath = new Path("/Spark_Top10NonUnique/input");
	private static Path outputPath = new Path("/Spark_Top10NonUnique/output");
	static int TOPN = 10;
	static String DIR = "asc";

	@SuppressWarnings("serial")
	public static void main(String[] args) {
		// 确保有正确的输入输出
		try {
			HDFSAPI hdfs = new HDFSAPI();
			hdfs.createDirectory(inputPath);
			hdfs.delDirectory(outputPath);
		} catch (IOException | InterruptedException | URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			System.out.println("输入输出文件路径有误");
		}
		// 创建于spark master的链接
		JavaSparkContext jsc = new JavaSparkContext();
		// 动态广播全局中的topN值 在任意一个节点都可获取到该变量值
		Broadcast<Integer> topN = jsc.broadcast(TOPN);
		// Broadcast<String> sorDir = jsc.broadcast(DIR);
		// 从hdfs读取输入文件并且创建第一个RDD
		JavaRDD<String> lines = jsc.textFile(inputPath.toString(), 1);
		lines.saveAsTextFile(outputPath.toString()+"/1");// 输出 读取的格式数据和结果做对比
		// RDD分区（重新将读取到的数据进行分区）
		JavaRDD<String> rdd = lines.coalesce(2);
		rdd.saveAsTextFile(outputPath.toString()+"/2");
		// 映射成键值对 会产重复的键（有可能）如（A分区有 【a,2 b,2 c,1 e,1】而B分区有【a,3,c5,f,10,g,4】
		// 则在各自的分区求得本地topN之后，再求总的topN时会出现两个a的top值，则是错误的表达）\
		// 1 先将各个分区中的数据，组成业务键值对
		JavaPairRDD<String, Integer> pairs = rdd.mapToPair(new PairFunction<String, String, Integer>() {

			@Override
			public Tuple2<String, Integer> call(String t) throws Exception {
				// TODO Auto-generated method stub
				String[] tokens = t.split(",");
				return new Tuple2<String, Integer>(tokens[0], Integer.valueOf(tokens[1]));
			}
		});
		pairs.saveAsTextFile(outputPath.toString()+"/3");
		// 然后归约重复的键 ，确保键是唯一的
		JavaPairRDD<String, Integer> uniquePairs = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {

			@Override
			public Integer call(Integer v1, Integer v2) throws Exception {
				// TODO Auto-generated method stub
				return v1 + v2;
			}
		});
		List<Tuple2<String, Integer>> tmp_keys = uniquePairs.collect();
		for (Tuple2<String, Integer> tuple2 : tmp_keys) {
			System.out.println("removed the repeat keys :  " + tuple2._1() + "--" + tuple2._2);
		}
		/*
		 * // 然后求得topN
		 * // 为各个输入分区创建一个本地top10 JavaRDD<SortedMap<Integer, String>> partitions =
		 * uniquePairs // 每次映射的是每个分区自己的数据 .mapPartitions(new
		 * FlatMapFunction<Iterator<Tuple2<String, Integer>>, SortedMap<Integer,
		 * String>>() {
		 * 
		 * @Override public Iterator<SortedMap<Integer, String>>
		 * call(Iterator<Tuple2<String, Integer>> t) throws Exception { // TODO
		 * Auto-generated method stub SortedMap<Integer, String> localTopN = new
		 * TreeMap<>(); while (t.hasNext()) { Tuple2<String, Integer> tup = t.next();
		 * localTopN.put(tup._2, tup._1); if (localTopN.size() > topN.getValue()) { if
		 * (sorDir.equals("asc")) localTopN.remove(localTopN.firstKey()); else
		 * localTopN.remove(localTopN.lastKey()); } } return
		 * Collections.singletonList(localTopN).iterator(); } }); //
		 * 收集所有本地的top10并且常见最终的top10列表 List<SortedMap<Integer, String>> allTopN =
		 * partitions.collect(); SortedMap<Integer, String> finalTopN = new TreeMap<>();
		 * for (SortedMap<Integer, String> t : allTopN) { for (Entry<Integer, String>
		 * entry : t.entrySet()) { System.out.println("这是在经过了本地topN排序之后的数据" +
		 * entry.getKey() + "--" + entry.getValue()); finalTopN.put(entry.getKey(),
		 * entry.getValue()); if (finalTopN.size() > topN.getValue()) { if
		 * (sorDir.equals("asc")) finalTopN.remove(finalTopN.firstKey()); else
		 * finalTopN.remove(finalTopN.lastKey()); } } } // 最终结果输出到文档 for (Entry<Integer,
		 * String> entry : finalTopN.entrySet()) { System.out.println("最终结果" +
		 * entry.getKey() + "=" + entry.getValue()); }
		 */
		// 然后求得topN
		// 使用 takeOrderd API来完成topN排序
		// 需要制定一个比较器，用来对键值对排序 一般自定义
		List<Tuple2<String, Integer>> finalTopN = uniquePairs.takeOrdered(topN.getValue(), new TopN_Tuple2Compartor());
		for (Tuple2<String, Integer> tuple2 : finalTopN) {
			System.out.println("最终结果" + tuple2._2 + "=" + tuple2._1);
		}
		jsc.close();
		System.exit(0);
	}
}
